package com.example.hzq;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Netty服务器，整合Disruptor实现高并发处理
 */
public class NettyDisruptorServer {
    private static final Logger logger = LoggerFactory.getLogger(NettyDisruptorServer.class);
    
    private final int port;
    private Disruptor<MessageEvent> disruptor;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    
    public NettyDisruptorServer(int port) {
        this.port = port;
    }
    
    public void start() throws Exception {
        // 配置SSL
        final SslContext sslCtx;
        if (false) { // 生产环境可开启SSL
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }
        
        // 初始化Disruptor
        int bufferSize = 1024 * 64;
        MessageEventFactory factory = new MessageEventFactory();
        ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        
        // 创建Disruptor
        disruptor = new Disruptor<>(
            factory,
            bufferSize,
            executor,
            ProducerType.MULTI,
            new YieldingWaitStrategy()
        );
        
        // 创建多个事件处理器，提高并行处理能力
        int handlerCount = Runtime.getRuntime().availableProcessors();
        MessageEventHandler[] handlers = new MessageEventHandler[handlerCount];
        for (int i = 0; i < handlerCount; i++) {
            handlers[i] = new MessageEventHandler(i);
        }
        
        // 连接处理器
        disruptor.handleEventsWithWorkerPool(handlers);
        
        // 启动Disruptor
        disruptor.start();
        
        // 获取RingBuffer用于发布事件
        final RingBuffer<MessageEvent> ringBuffer = disruptor.getRingBuffer();
        
        // 配置Netty
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 1024)
             .option(ChannelOption.SO_REUSEADDR, true)
             .option(ChannelOption.TCP_NODELAY, true)
             .option(ChannelOption.SO_KEEPALIVE, true)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     if (sslCtx != null) {
                         ch.pipeline().addLast(sslCtx.newHandler(ch.alloc()));
                     }
                     ch.pipeline().addLast(
                         new StringDecoder(),
                         new StringEncoder(),
                         new DisruptorHandler(ringBuffer)
                     );
                 }
             });
            
            // 启动服务器
            ChannelFuture f = b.bind(port).sync();
            logger.info("Server started on port: {}", port);
            
            // 等待服务器关闭
            f.channel().closeFuture().sync();
        } finally {
            stop();
        }
    }
    
    public void stop() {
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
        }
        if (workerGroup != null) {
            workerGroup.shutdownGracefully();
        }
        if (disruptor != null) {
            disruptor.shutdown();
        }
        logger.info("Server stopped");
    }
    
    public static void main(String[] args) throws Exception {
        int port = 8080;
        if (args.length > 0) {
            port = Integer.parseInt(args[0]);
        }
        new NettyDisruptorServer(port).start();
    }
}
